package com.insight.common.mq;

import io.lettuce.core.api.async.RedisAsyncCommands;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Slf4j
public class AckFlusher {

    private final RedisAsyncCommands<String, String> async;
    private final String stream;
    private final String group;

    private final BlockingQueue<String> ackQueue = new LinkedBlockingQueue<>();
    private final int ackBatchSize;
    private final long ackMaxIntervalMs;

    public AckFlusher(RedisAsyncCommands<String, String> async,
                      String stream, String group,
                      int ackBatchSize, long ackMaxIntervalMs) {
        this.async = async;
        this.stream = stream;
        this.group = group;
        this.ackBatchSize = ackBatchSize;
        this.ackMaxIntervalMs = ackMaxIntervalMs;

        Executors.newSingleThreadExecutor().submit(this::flushLoop);
    }

    public void addAckId(String ackId) {
        ackQueue.offer(ackId);
    }

    private void flushLoop() {
        List<String> buffer = new ArrayList<>();
        long lastFlushTime = System.currentTimeMillis();

        while (true) {
            try {
                String ackId = ackQueue.poll(1, TimeUnit.MILLISECONDS);
                if (ackId != null) {
                    buffer.add(ackId);
                }
                boolean shouldFlush = false;

                if (buffer.size() >= ackBatchSize) {
                    shouldFlush = true;
                } else if (!buffer.isEmpty() && System.currentTimeMillis() - lastFlushTime >= ackMaxIntervalMs) {
                    shouldFlush = true;
                }

                if (shouldFlush) {
                    flushNow(buffer);
                    lastFlushTime = System.currentTimeMillis();
                }

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.error("[{}/{}] flushLoop 被打断", stream, group);
                log.error(e.getMessage(), e);
                break;
            }
        }
    }

    private void flushNow(List<String> buffer) {
        if (buffer.isEmpty()) return;

        List<String> toAck = new ArrayList<>(buffer);
        buffer.clear();

        async.xack(stream, group, toAck.toArray(new String[0]))
                .thenAccept(count -> {
                    log.info("[{}/{}] 异步 ACK {} 条", stream, group, count);
                })
                .exceptionally(ex -> {
                    log.error("[{}/{}] ACK 失败：{}", stream, group, ex.getMessage());
                    log.error(ex.getMessage(), ex);
                    return null;
                });
    }
}
